package cn.aposoft.mq.kafka.admin.topic;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartition;

import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * PartitionReassignments :重新定义Partition的Replica的集合
 *
 * @param TopicPartition：待调整的Topic分区
 * @param NewPartitionReassignment:  brokerlist，集合中的broker会被分配副本，不在集合中的broker会删除已有的副本
 */
public class AlterPartitionReassignments {
    public static void main(String[] args) {
        Properties p = new Properties();
        String topicName = "test-new";
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "aposoft.cn:10091,aposoft.cn:10092,aposoft.cn:10093");// kafka地址，多个地址用逗号分割
        try (Admin admin = Admin.create(p);) {
            TopicPartition partition = new TopicPartition(topicName, 0);
            NewPartitionReassignment newPartitionReassignment = new NewPartitionReassignment(Collections.singletonList(92));
            Map<TopicPartition, Optional<NewPartitionReassignment>> reassignments = new HashMap<>();
            reassignments.put(partition, Optional.of(newPartitionReassignment));
            AlterPartitionReassignmentsResult result = admin.alterPartitionReassignments(reassignments);
            ListPartitionReassignmentsResult resultList = admin.listPartitionReassignments();
            KafkaFuture<Map<TopicPartition, PartitionReassignment>> resultListFuture = resultList.reassignments();
            System.out.println(resultListFuture.get());
            KafkaFuture<Void> future = result.all();
            future.get();
            Map<TopicPartition, KafkaFuture<Void>> valueFuture = result.values();
            System.out.println(valueFuture.get(partition));
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
